package com.cw.base.common.dispatch.queue;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.PriorityBlockingQueue;

import com.cw.base.common.dispatch.event.EventWrapper;
import com.cw.base.common.dispatch.event.MessageEvent;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;

import com.cw.base.common.dispatch.pojo.Task;
import com.cw.base.common.dispatch.utils.MapDbUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * 对Q的封装
 * Created on 15/11/9.
 *
 * @author zhangcheng
 */
@Slf4j
public class TaskQueue implements Queue, InitializingBean {

    /** 线程用到的锁对象 */
    private final Object object = new Object();
    private PriorityBlockingQueue<Task> taskBlockingQueue;
    /** 程序是否已关闭 */
    private boolean close = false;

    private String fileName;

    @Autowired
    @Qualifier("eventWrapper")
    private EventWrapper<MessageEvent> wrapper;

    public TaskQueue(int queueSize) {
        this.taskBlockingQueue = new PriorityBlockingQueue<>(queueSize);
        init(null);
    }

    public TaskQueue(int queueSize, String fileName) {
        this.taskBlockingQueue = new PriorityBlockingQueue<>(queueSize);
        init(fileName);
    }

    private void init(String fileName) {
        this.fileName = fileName;
        Runtime.getRuntime()
            .addShutdownHook(new Thread(() -> {
                try {
                    Thread.sleep(5000);
                    close = true;
                } catch (InterruptedException e) {
                    log.error("线程处理异常");
                }
            }));
    }

    @Override
    public void put(final Task task) {
        // 信号量用来限流
        if (!close) {
            this.taskBlockingQueue.put(task);
            List<Task> tasks = new ArrayList<>();
            tasks.add(task);
            MapDbUtil.addData(tasks);
            wrapper.post(new MessageEvent(this::getFirst));
        } else {
            throw new RuntimeException("关闭了队列，JVM 将关闭");
        }
    }

    @Override
    public void remove(String key) {
        log.debug("删除操作");
    }

    @Override
    public Task getFirst() {
        Task task = null;
        try {
            task = this.taskBlockingQueue.take();
        } catch (InterruptedException e) {
            log.error("线程处理异常");
        }
        return task;
    }

    @Override
    public void afterPropertiesSet() {
        //读取信息到queue里面,删除queue里面的信息
        List<Task> taskList = MapDbUtil.fetchTask(fileName);
        for (Task task : taskList) {
            put(task);
        }
    }

    /**
     * 持久化对象线程
     */
    private class PersistenceQueueThread extends Thread {

        @Override
        public void run() {
            try {
                while (true) {
                    synchronized (object) {
                        if (taskBlockingQueue.size() <= 0) {
                            // 队列是空得，则休眠100毫秒
                            object.wait();
                        }
                    }
                }
            } catch (InterruptedException e) {
                log.error("线程处理异常");
                Thread.currentThread()
                    .interrupt();
            }
        }

        public void doNotify() {
            synchronized (object) {
                object.notifyAll();
            }
        }

    }

}
